package org.apache.storm.starter.trident;
/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 * Contains some contributions under the Thrift Software License.
 * Please see doc/old-thrift-license.txt in the Thrift distribution for
 * details.
 */

import java.util.Properties;

import org.apache.storm.starter.spout.RandomSentenceSpout;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import storm.trident.Stream;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.FilterNull;
import storm.trident.operation.builtin.MapGet;
import storm.trident.testing.MemoryMapState;
import storm.trident.testing.Split;

/**
 * A sample word count trident topology using transactional kafka spout that has
 * the following components.
 * <ol>
 * <li>{@link KafkaBolt}
 * that receives random sentences from {@link RandomSentenceSpout} and
 * publishes the sentences to a kafka "test" topic.
 * </li>
 * <li>{@link TransactionalTridentKafkaSpout}
 * that consumes sentences from the "test" topic, splits it into words,
 * aggregates
 * and stores the word count in a {@link MemoryMapState}.
 * </li>
 * <li>DRPC query
 * that returns the word counts by querying the trident state (MemoryMapState).
 * </li>
 * </ol>
 * <p>
 * For more background read the
 * <a href="https://storm.apache.org/documentation/Trident-tutorial.html">
 * trident tutorial</a>,
 * <a href="https://storm.apache.org/documentation/Trident-state">trident
 * state</a> and
 * <a href="https://github.com/apache/storm/tree/master/external/storm-kafka">
 * Storm Kafka </a>.
 * </p>
 */
public class TridentKafkaWordCount {

//    private String zkUrl;
//    private String brokerUrl;
//
//    TridentKafkaWordCount(String zkUrl, String brokerUrl) {
//        this.zkUrl = zkUrl;
//        this.brokerUrl = brokerUrl;
//    }
//
//    /**
//     * Creates a transactional kafka spout that consumes any new data published to "test" topic.
//     * <p/>
//     * For more info on transactional spouts
//     * see "Transactional spouts" section in
//     * <a href="https://storm.apache.org/documentation/Trident-state"> Trident state</a> doc.
//     *
//     * @return a transactional trident kafka spout.
//     */
//    private TransactionalTridentKafkaSpout createKafkaSpout() {
//        ZkHosts hosts = new ZkHosts(zkUrl);
//        TridentKafkaConfig config = new TridentKafkaConfig(hosts, "test");
//        config.scheme = new SchemeAsMultiScheme(new StringScheme());
//
//        // Consume new data from the topic
//        config.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
//        return new TransactionalTridentKafkaSpout(config);
//    }
//
//
//    private Stream addDRPCStream(TridentTopology tridentTopology, TridentState state, LocalDRPC drpc) {
//        return tridentTopology.newDRPCStream("words", drpc)
//                .each(new Fields("args"), new Split(), new Fields("word"))
//                .groupBy(new Fields("word"))
//                .stateQuery(state, new Fields("word"), new MapGet(), new Fields("count"))
//                .each(new Fields("count"), new FilterNull())
//                .project(new Fields("word", "count"));
//    }
//
//    private TridentState addTridentState(TridentTopology tridentTopology) {
//        return tridentTopology.newStream("spout1", createKafkaSpout()).parallelismHint(1)
//                .each(new Fields("str"), new Split(), new Fields("word"))
//                .groupBy(new Fields("word"))
//                .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
//                .parallelismHint(1);
//    }
//
//    /**
//     * Creates a trident topology that consumes sentences from the kafka "test" topic using a
//     * {@link TransactionalTridentKafkaSpout} computes the word count and stores it in a {@link MemoryMapState}.
//     * A DRPC stream is then created to query the word counts.
//     * @param drpc
//     * @return
//     */
//    public StormTopology buildConsumerTopology(LocalDRPC drpc) {
//        TridentTopology tridentTopology = new TridentTopology();
//        addDRPCStream(tridentTopology, addTridentState(tridentTopology), drpc);
//        return tridentTopology.build();
//    }
//
//    /**
//     * Return the consumer topology config.
//     *
//     * @return the topology config
//     */
//    public Config getConsumerConfig() {
//        Config conf = new Config();
//        conf.setMaxSpoutPending(20);
//        //  conf.setDebug(true);
//        return conf;
//    }
//
//    /**
//     * A topology that produces random sentences using {@link RandomSentenceSpout} and
//     * publishes the sentences using a KafkaBolt to kafka "test" topic.
//     *
//     * @return the storm topology
//     */
//    public StormTopology buildProducerTopology(Properties prop) {
//        TopologyBuilder builder = new TopologyBuilder();
//        builder.setSpout("spout", new RandomSentenceSpout(), 2);
//        /**
//         * The output field of the RandomSentenceSpout ("word") is provided as the boltMessageField
//         * so that this gets written out as the message in the kafka topic.
//         */
//        KafkaBolt bolt = new KafkaBolt().withProducerProperties(prop)
//                .withTopicSelector(new DefaultTopicSelector("test"))
//                .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("key", "word"));
//        builder.setBolt("forwardToKafka", bolt, 1).shuffleGrouping("spout");
//        return builder.createTopology();
//    }
//
//    /**
//     * Returns the storm config for the topology that publishes sentences to kafka "test" topic using a kafka bolt.
//     * The KAFKA_BROKER_PROPERTIES is needed for the KafkaBolt.
//     *
//     * @return the topology config
//     */
//    public Properties getProducerConfig() {
//        Properties props = new Properties();
//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
//        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//        props.put(ProducerConfig.CLIENT_ID_CONFIG, "storm-kafka-producer");
//        return props;
//    }
//
//    /**
//     * <p>
//     * To run this topology ensure you have a kafka broker running.
//     * </p>
//     * Create a topic test with command line,
//     * kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition 1 --topic test
//     */
//    public static void main(String[] args) throws Exception {
//
//        String zkUrl = "localhost:2181";        // the defaults.
//        String brokerUrl = "localhost:9092";
//
//        if (args.length > 2 || (args.length == 1 && args[0].matches("^-h|--help$"))) {
//            System.out.println("Usage: TridentKafkaWordCount [kafka zookeeper url] [kafka broker url]");
//            System.out.println("   E.g TridentKafkaWordCount [" + zkUrl + "]" + " [" + brokerUrl + "]");
//            System.exit(1);
//        } else if (args.length == 1) {
//            zkUrl = args[0];
//        } else if (args.length == 2) {
//            zkUrl = args[0];
//            brokerUrl = args[1];
//        }
//
//        System.out.println("Using Kafka zookeeper url: " + zkUrl + " broker url: " + brokerUrl);
//
//        TridentKafkaWordCount wordCount = new TridentKafkaWordCount(zkUrl, brokerUrl);
//
//        LocalDRPC drpc = new LocalDRPC();
//        LocalCluster cluster = new LocalCluster();
//
//        // submit the consumer topology.
//        cluster.submitTopology("wordCounter", wordCount.getConsumerConfig(), wordCount.buildConsumerTopology(drpc));
//
//        Config conf = new Config();
//        conf.setMaxSpoutPending(20);
//        // submit the producer topology.
//        cluster.submitTopology("kafkaBolt", conf, wordCount.buildProducerTopology(wordCount.getProducerConfig()));
//
//        // keep querying the word counts for a minute.
//        for (int i = 0; i < 60; i++) {
//            System.out.println("DRPC RESULT: " + drpc.execute("words", "the and apple snow jumped"));
//            Thread.sleep(1000);
//        }
//
//        cluster.killTopology("kafkaBolt");
//        cluster.killTopology("wordCounter");
//        cluster.shutdown();
//    }
}
